package app.dws;

import Bean.UserComeBackBean;
import app.dwd.BaseAppV1;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import common.Constant;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import util.AtguiguUtil;
import util.FlinkSinkUtil;

import java.time.Duration;

/**
 * @author cheng
 * @create 2022-10-26 14:39
 */
public class Dws_UserComeBack extends BaseAppV1 {
    public static void main(String[] args) {
        new Dws_UserComeBack().init(4010,
                2,
                "Dws_UserComeBack",
                Constant.TOPIC_DWD_TRAFFIC_PAGE);
    }

    @Override
    protected void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {
        //1.过滤page
        SingleOutputStreamOperator<JSONObject> comebackFromPageStream = findComebackFromPage(stream);

        //2.转成pojo类型
        SingleOutputStreamOperator<UserComeBackBean> streamOperator = pareToObject(comebackFromPageStream);
//streamOperator.print();
        //3.开窗聚合
        SingleOutputStreamOperator<UserComeBackBean> windowAggStream = windowAgg(streamOperator);

        //4.写出到clickhouse
        writeToClickHouse(windowAggStream);

    }

    private void writeToClickHouse(SingleOutputStreamOperator<UserComeBackBean> windowAggStream) {
        windowAggStream.addSink(FlinkSinkUtil.getClickHouseSink("dws_user_user_login_window", UserComeBackBean.class));

    }

    private SingleOutputStreamOperator<UserComeBackBean> windowAgg(SingleOutputStreamOperator<UserComeBackBean> streamOperator) {
        SingleOutputStreamOperator<UserComeBackBean> aggregate = streamOperator
                .windowAll(TumblingEventTimeWindows.of(Time.seconds(5)))
                .aggregate(new AggregateFunction<UserComeBackBean, Tuple3<Long,Long,Long>, UserComeBackBean>() {
                               @Override
                               public Tuple3<Long, Long, Long> createAccumulator() {
                                   return new Tuple3<Long, Long, Long>(0L,0L,0L);
                               }

                               @Override
                               public Tuple3<Long, Long, Long> add(UserComeBackBean userComeBackBean, Tuple3<Long, Long, Long> longLongLongTuple3) {
                                   return new Tuple3<Long, Long, Long>(userComeBackBean.getBackCt() + longLongLongTuple3.f0,userComeBackBean.getUuCt() + longLongLongTuple3.f1 , userComeBackBean.getAddCt() + longLongLongTuple3.f2);
                               }

                               @Override
                               public UserComeBackBean getResult(Tuple3<Long, Long, Long> longLongLongTuple3) {
                                   return new UserComeBackBean("","",longLongLongTuple3.f0,longLongLongTuple3.f1,longLongLongTuple3.f2,System.currentTimeMillis());
                               }

                               @Override
                               public Tuple3<Long, Long, Long> merge(Tuple3<Long, Long, Long> longLongLongTuple3, Tuple3<Long, Long, Long> acc1) {
                                   return null;
                               }
                           }

                        , new ProcessAllWindowFunction<UserComeBackBean, UserComeBackBean, TimeWindow>() {
                            @Override
                            public void process(Context context, Iterable<UserComeBackBean> iterable, Collector<UserComeBackBean> collector) throws Exception {
                                UserComeBackBean bean = iterable.iterator().next();

                                bean.setStt(AtguiguUtil.toDatTime(context.window().getStart()));
                                bean.setEdt(AtguiguUtil.toDatTime(context.window().getEnd()));

                                collector.collect(bean);
                            }
                        });
        return aggregate;
    }

    private SingleOutputStreamOperator<UserComeBackBean> pareToObject(SingleOutputStreamOperator<JSONObject> comebackFromPageStream) {
return comebackFromPageStream
        .assignTimestampsAndWatermarks(
                WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(3))
        .withTimestampAssigner((obj,ts) -> obj.getLong("ts"))
        .withIdleness(Duration.ofSeconds(5))
        )
        .keyBy(obj -> obj.getJSONObject("common").getString("uid"))
        .process(new KeyedProcessFunction<String, JSONObject, UserComeBackBean>() {

         private  ValueState<String> lastVisitDateState ;
            @Override
            public void open(Configuration parameters) throws Exception {
                 lastVisitDateState = getRuntimeContext().getState(new ValueStateDescriptor<String>("lastVisitDateState", String.class));
            }

            @Override
            public void processElement(JSONObject jsonObject, Context context, Collector<UserComeBackBean> collector) throws Exception {
                Long ts = jsonObject.getLong("ts");


                long backCt = 0L;
                long uuCt = 0L;
                long addCt = 0L;
                //
                String today = AtguiguUtil.toDate(ts);
                String lastVisitDate = lastVisitDateState.value();
                System.out.println("today  " +today);
                System.out.println("last " + lastVisitDate);
                if (!today.equals(lastVisitDate)) {
                    // 今天的第一次登录
                    if(lastVisitDate == null){
                        addCt = 1L;
                    }

                    uuCt = 1L;
                    lastVisitDateState.update(today);
//                    lastVisitDate = lastVisitDateState.value();
                    System.out.println("==========");
                    // 需要判断是否为7 日回流用户
                    if (lastVisitDate != null) {
                        // 证明以前登录过, 需要判断是否为 回流
                        if ((ts - AtguiguUtil.dateToTs(lastVisitDate)) / 1000 / 60 / 60 / 24 >= 7) {
//                        if(true){
                            System.out.println("7d");
                            backCt = 1L;
                        }
                    }
                }
                UserComeBackBean bean = new UserComeBackBean("", "",
                        backCt,
                        uuCt,
                        addCt,
                        ts
                );

                if (uuCt == 1L) {
                    collector.collect(bean);
                }
            }
        });
        
        
    }

    private SingleOutputStreamOperator<JSONObject> findComebackFromPage(DataStreamSource<String> stream) {

        return stream
                .map(JSON::parseObject)
                .filter(new FilterFunction<JSONObject>() {
                    @Override
                    public boolean filter(JSONObject jsonObject) throws Exception {
                        String uid = jsonObject.getJSONObject("common").getString("uid");
                        String page = jsonObject.getJSONObject("page").getString("page_id");

                        return uid != null && page != null;
                    }
                });
    }
}
